package tkM

import (
	"bytes"
	"fmt"
	ea "db2s/encryptionAlgorithm"
	"db2s/global"
	"db2s/go-log/log"
	"db2s/outPut"
	"db2s/ref"
	mq "db2s/topic-mq"
	"reflect"
	"strings"
	"sync/atomic"
)

func errorActive(plan *SchedulePlan2, err string) (errStr string) {
	if strings.EqualFold(err, "column length abnormal") { //表结构不一致
		var srcColumn, dstColumn []string
		var returnError bytes.Buffer
		for _, v := range plan.Object.TableColData.SColumnInfo {
			srcColumn = append(srcColumn, v.ColumnName)
		}
		for _, v := range plan.Object.TableColData.DColumnInfo {
			dstColumn = append(dstColumn, v.ColumnName)
		}
		aa := &ea.CheckSumTypeStruct{}
		add, del := aa.Arrcmp(srcColumn, dstColumn)
		for _, v := range add {
			returnError.WriteString(fmt.Sprintf("add column %v ,", v))
		}
		for _, v := range del {
			returnError.WriteString(fmt.Sprintf("drop column %v,", v))
		}
		errStr = returnError.String()
		returnError.Reset()
	} else {
		errStr = err
	}
	return
}
func initErrorAction(plan *SchedulePlan2, err error, podInfo string) {
	atomic.AddInt64(plan.Status.ErrEnd, -1)
	PlanContext.TaskBarSubsErrorSwap(plan.subTaskInfo, errorActive(plan, fmt.Sprintf("%v", err)))
	if PlanContext.TaskBarSubsStatusGet(plan.subTaskInfo) != -2 {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
	}
	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
}
func runErrorAction(message mq.Message, err error) {
	var plan = message.Properties["plan"].(*SchedulePlan2)
	if err == nil {
		return
	}
	PlanContext.TPodTableStatusSwap(plan.TPod.Status, "error")
	atomic.SwapInt64(plan.MQ.ErrStatus, -1)
	atomic.SwapInt64(plan.Status.ErrEnd, -1)
	initErrorAction(plan, err, "errors")
	return
}
func abnormalErrorQuit(plan *SchedulePlan2) bool {
	return atomic.LoadInt64(plan.Status.ErrEnd) < 0
}
func handleError(message mq.Message, event string, err error) {
	errMsg := ref.ErrAddPrintf(event, err)
	runErrorAction(message, errMsg)
	log.ErrorLog().Error(fmt.Sprintf("%v", errMsg))
}
func SendMsgExecSumFunc(param any) global.EFunc {
	return global.EFunc{
		FuncName: func(event string, s any) (r any) {
			var point []any
			var val int64
			switch s.(type) {
			case []any:
				for k, v := range s.([]any) {
					switch k {
					case 0:
						switch reflect.TypeOf(v).String() {
						case "interface {}", "*int64":
							point = append(point, v)
						case "[]interface {}":
							for _, q := range v.([]any) {
								point = append(point, q)
							}
						}
					case 1:
						val = v.(int64)
					}
				}
			}
			for _, v := range point {
				switch v.(type) {
				case *int64:
					atomic.AddInt64(v.(*int64), val)
				}
			}
			return
		},
		Params: param,
	}
}
func terminalOutputTPod(plan *SchedulePlan2) {
	if e := atomic.LoadInt64(plan.Status.ErrEnd); e < 0 {
		plan.TPod.Result = "error"
	}
	plan.TPod.Seq = plan.Object.TaskSeq
	newTer.Write(plan.TPod)
}
func mqStatusMonitorAbnormalQuit(event string, i any) (r any) {
	var plan *SchedulePlan2
	switch i.(type) {
	case *SchedulePlan2:
		plan = i.(*SchedulePlan2)
	default:
		return
	}
	terminalOutputTPod(plan)
	if err := xlsWriterTerminalData(plan); err != nil {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
		return
	}
	PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "error")
	return
}
func writeConfigFileText(plan *SchedulePlan2) {
	var event = "[writeConfigFileText]"
	for subMode, Point := range generalConfigText {
		subFile := taskNameGenerate(subMode, plan)
		for _, subPoint := range Point {
			var fileName = fmt.Sprintf("%v", subPoint.Print(outPut.FileName))
			if !strings.HasSuffix(fileName, subFile) {
				continue
			}
			if err := subPoint.Write("", generalConfigObjectProperty(subMode, plan)); err != nil {
				log.MainLog().Warn(fmt.Sprintf("(%d) %v gtConfig write %v fail for table %v.%v. error is {%v}", plan.Object.ObjectLogSeq, event, fileName, plan.Object.Schema, plan.Object.Table, err))
			}
		}
	}
}
func mqStatusMonitorNormalQuit(event string, i any) (r any) {
	var plan *SchedulePlan2
	defer func() {
		PlanContext.TaskBarSubsStatusSwap(plan.subTaskInfo, "finish")
	}()
	switch i.(type) {
	case *SchedulePlan2:
		plan = i.(*SchedulePlan2)
	default:
		fmt.Println("---------errr:", reflect.TypeOf(i))
		return
	}
	event = "[mqStatusMonitorNormalQuit]"
	log.MainLog().Debug(fmt.Sprintf("(%d) %v It is detected that the message queue production quantity and consumption quantity in table %v.%v are equal.", plan.Object.ObjectLogSeq, event, plan.Object.Schema, plan.Object.Table))
	terminalOutputTPod(plan)
	writeConfigFileText(plan)
	if err := xlsWriterTerminalData(plan); err != nil {
		return
	}
	log.MainLog().Info(fmt.Sprintf("(%v) %v Table %v.%v data migration is completed.", plan.Object.ObjectLogSeq, event, plan.Object.Schema, plan.Object.Table))
	return
}

func getMsgSendData(plan *SchedulePlan2, si map[string]any) (res map[string]any) {
	res = map[string]any{
		"plan":    plan,
		"check":   getCheckMod(),
		"subTask": plan.subTaskInfo,
		"err":     plan.Status.ErrEnd,
		"outFay":  GlobalPConfigs.result.Teletypewriter,
	}
	for k, v := range si {
		res[k] = v
	}
	return
}
